package org.apache.hbase.example;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HbaseCrud {
	
	//新建表
    public static boolean create(String tableName, String columnFamily)
            throws Exception {
    	Connection connection = HbaseUtil.getConnection();
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();

        if (admin.tableExists(TableName.valueOf(tableName))) {
            System.out.println(tableName + " exists!");
            return false;
        } else {
            String[] columnFamilyArray = columnFamily.split(",");
            List<ColumnFamilyDescriptor> cf = new ArrayList<>();
            for (int i = 0; i < columnFamilyArray.length; i++) {
                cf.add(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamilyArray[i])).setMaxVersions(3).build());
            }
            TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
            tableBuilder.setColumnFamilies(cf);
            
            admin.createTable(tableBuilder.build());
            System.out.println(tableName + " create successfully!");
            return true;
        }
    }
 
    //插入数据
    public static boolean put(String tablename, String row, String columnFamily,
                              String qualifier, String data) throws Exception {
    	Connection connection = HbaseUtil.getConnection();
        Table table = connection.getTable(TableName.valueOf(tablename));
    
        Put put = new Put(Bytes.toBytes(row));
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier),
                Bytes.toBytes(data));
        table.put(put);
        System.out.println("put '" + row + "', '" + columnFamily + ":" + qualifier
                + "', '" + data + "'");
        return true;
    }
    
    //查看某行
    public static String get(String tablename, String row, String cf, String cq) throws Exception {
    	Connection connection = HbaseUtil.getConnection();
        Table table = connection.getTable(TableName.valueOf(tablename));
        Get get = new Get(Bytes.toBytes(row)).readAllVersions();

        Result result = table.get(get);
        List<Cell> cells = result.getColumnCells(Bytes.toBytes(cf), Bytes.toBytes(cq));
        for(Cell cell : cells){
        	byte[] value = new byte[cell.getValueLength()];
        	System.arraycopy(cell.getValueArray(), cell.getValueOffset(),value , 0, cell.getValueLength());
        	System.out.println(cell.getTimestamp()+":"+Bytes.toString(value));
        }
       
        System.out.println(Bytes.toString(result.getValue(Bytes.toBytes(cf), Bytes.toBytes(cq))));
        return result.toString();
    }
    
    public static void scan(String tablename, String f1, String q1) throws Exception {
    	Connection connection = HbaseUtil.getConnection();
        Table table = connection.getTable(TableName.valueOf(tablename));
        Scan s = new Scan().withStartRow(Bytes.toBytes("key1")).withStopRow(Bytes.toBytes("key4"));
        s.setCaching(5);
       
        Filter filter = new ValueFilter(CompareOperator.EQUAL, new RegexStringComparator(".*试啊3"));
        s.setFilter(filter);
        ResultScanner rs = table.getScanner(s);
       
        for (Result r : rs) {
        	System.out.println(Bytes.toString(r.getRow())+":"+Bytes.toString(r.getValue(Bytes.toBytes(f1), Bytes.toBytes(q1))));
        }
    }
    
    //删除表
    public static boolean delete(String tableName) throws IOException {
    	Connection connection = HbaseUtil.getConnection();
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        TableName tn = TableName.valueOf(tableName);
        if (admin.tableExists(tn)) {
            try {
                admin.disableTable(tn);
                admin.deleteTable(tn);
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }
        return true;
    }
 
    //删除ColumnFamily
    public static boolean deleteColumnFamily(String tableName,String columnFamilyName) throws IOException {
    	Connection connection = HbaseUtil.getConnection();
    	HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        TableName tn = TableName.valueOf(tableName);
        if (admin.tableExists(tn)) {
            try {
                admin.deleteColumnFamily(tn,Bytes.toBytes(columnFamilyName));
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }
        return true;
    }
    //删除row
    public static boolean deleteRow(String tableName,String rowName) throws IOException {
    	Connection connection = HbaseUtil.getConnection();
    	HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        Table table = connection.getTable(TableName.valueOf(tableName));
        if (admin.tableExists(TableName.valueOf(tableName))) {
            try {
                Delete delete = new Delete(rowName.getBytes());
                table.delete(delete);
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }
        return true;
    }
 
    //删除qualifier
    public static boolean deleteQualifier(String tableName,String rowName,String columnFamilyName,String qualifierName) throws IOException {
    	Connection connection = HbaseUtil.getConnection();
    	HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
        Table table = connection.getTable(TableName.valueOf(tableName));
        if (admin.tableExists(TableName.valueOf(tableName))) {
            try {
                Delete delete = new Delete(rowName.getBytes());
                delete.addColumns(columnFamilyName.getBytes(),qualifierName.getBytes());
                table.delete(delete);
            } catch (Exception e) {
                e.printStackTrace();
                return false;
            }
        }
        return true;
    }
    public static void main(String[] args) throws Exception {
		create("htest", "cf1,cf2");
    	
    	put("htest", "key3", "cf1", "cq1", "我的测试啊3");
    	put("htest", "key4", "cf1", "cq1", "我的测试啊4");
    	put("htest", "key5", "cf1", "cq1", "我的测试啊5");
    	get("htest", "key1", "cf1", "cq1");
    	scan("htest",  "cf1", "cq1");
    	//deleteQualifier("htest", "key1","cf1", "cq1" );
    	//deleteRow("htest", "key1");
    	// deleteColumnFamily("htest", "cf1");
    	//delete("htest");
	}

}
